package com.example.demo.manager.publisher;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.example.demo.manager.inf.CanalManager;
import com.example.demo.utils.CanalPropertyUtils;

/**
 * @Description
 * @Author <a href="hedong.wu@baozun.com">wuhedong</a>
 * @Date 2020/5/26 16:38
 **/
public interface CanalChangePublisher<T> {

    /**
     * 默认推送
     *
     * @param entry canal的数据
     */
    default void publish(CanalEntry.Entry entry) {
        CanalEntry.RowChange rowChage;
        try {
            rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
        } catch (Exception e) {
            throw new RuntimeException("ERROR ## parser of eromanga-event has an error , " + "data:" + entry.toString(), e);
        }

        CanalManager<T> canalManager = getCanalManager(entry.getHeader().getTableName());
        if (canalManager == null) {
            return;
        }

        CanalEntry.EventType eventType = rowChage.getEventType();
        doEntrys(rowChage, canalManager, eventType);
    }

    /**
     * 处理数据
     *
     * @param rowChage     行数据
     * @param canalManager 处理数据的类
     * @param eventType    数据类型
     */
    void doEntrys(CanalEntry.RowChange rowChage, CanalManager<T> canalManager, CanalEntry.EventType eventType);

    /**
     * 返回处理的类  根据表名
     *
     * @param tableName 表名
     * @return 返回的处理类
     */
    CanalManager<T> getCanalManager(String tableName);

    /**
     * 默认的处理单行数据方法
     *
     * @param eventType    时间类型 增删改
     * @param canalManager 处理数据的manager
     * @param rowData      数据
     * @return 处理结果
     */
    default boolean handle(CanalEntry.EventType eventType, CanalManager<T> canalManager, CanalEntry.RowData rowData) {
        if (eventType == CanalEntry.EventType.DELETE) {
            T message = CanalPropertyUtils.makeDomain(rowData.getBeforeColumnsList(), canalManager.getHandleClass(), false);
            return !canalManager.doOnDelete(message);
        } else if (eventType == CanalEntry.EventType.INSERT) {
            T message = CanalPropertyUtils.makeDomain(rowData.getAfterColumnsList(), canalManager.getHandleClass(), false);
            return !canalManager.doOnAdd(message);
        } else {
            T beforemessage = CanalPropertyUtils.makeDomain(rowData.getBeforeColumnsList(), canalManager.getHandleClass(), false);
            T aftermessage = CanalPropertyUtils.makeDomain(rowData.getAfterColumnsList(), canalManager.getHandleClass(), false);
            return !canalManager.doOnUpdate(beforemessage, aftermessage);
        }
    }

}
